package com.esgyn;
import com.esgyn.protoInterface.Sum;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.coprocessor.example.generated.ExampleProtos;
import org.apache.hadoop.hbase.ipc.CoprocessorRpcUtils;
import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;

import java.io.IOException;
import java.util.Map;

public class TestRowCountEndpoint {
    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        // conf.set("hbase.zookeeper.quorum", "flink1,flink2,flink3");
        Connection connection = ConnectionFactory.createConnection(conf);
        TableName tableName = TableName.valueOf("users");
        Table table = connection.getTable(tableName);

        final ExampleProtos.CountRequest request = ExampleProtos.CountRequest.newBuilder().build();
        try {
            Map<byte[], Long> results = table.coprocessorService(
                    ExampleProtos.RowCountService.class,
                    null,  /* start key */
                    null,  /* end   key */
                    new Batch.Call<ExampleProtos.RowCountService, Long>() {
                        @Override
                        public Long call(ExampleProtos.RowCountService rowCountService) throws IOException {
                            CoprocessorRpcUtils.BlockingRpcCallback<ExampleProtos.CountResponse> rpcCallback = new CoprocessorRpcUtils.BlockingRpcCallback<>();
                            rowCountService.getRowCount(null, request, rpcCallback);
                            ExampleProtos.CountResponse response = rpcCallback.get();
                            return response.hasCount() ? response.getCount() : 0L;
                        }
                    }
            );

            for (Long count : results.values()) {
                System.out.println("rowCount = " + count);
            }
        } catch (ServiceException e) {
            e.printStackTrace();
        } catch (Throwable e) {
            e.printStackTrace();
        }


//        final Sum.SumRequest request = Sum.SumRequest.newBuilder().setFamily("salaryDet").setColumn("gross").build();
//        try {
//            Map<byte[], Long> results = table.coprocessorService(
//                    Sum.SumService.class,
//                    null,  /* start key */
//                    null,  /* end   key */
//                    new Batch.Call<Sum.SumService, Long>() {
//                        @Override
//                        public Long call(Sum.SumService aggregate) throws IOException {
//                            CoprocessorRpcUtils.BlockingRpcCallback<Sum.SumResponse> rpcCallback = new CoprocessorRpcUtils.BlockingRpcCallback<>();
//                            aggregate.getSum(null, request, rpcCallback);
//                            Sum.SumResponse response = rpcCallback.get();
//
//                            return response.hasSum() ? response.getSum() : 0L;
//                        }
//                    }
//            );
//
//            for (Long sum : results.values()) {
//                System.out.println("Sum = " + sum);
//            }
//        } catch (ServiceException e) {
//            e.printStackTrace();
//        } catch (Throwable e) {
//            e.printStackTrace();
//        }
    }
}